/** https://github.com/apache/rocketmq-clients/blob/golang/v5.1.3/golang/example/consumer/push_consumer/main.go */
package main

import (
	"fmt"
	"gitee.com/zackeus/go-zero/core/logx"
	"os"
	"time"

	rmq_client "github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
)

const (
	Topic         = "test-topic"
	ConsumerGroup = "test-group"
	// Endpoint 多个封号分割
	Endpoint  = "192.168.137.31:8081"
	NameSpace = ""
	AccessKey = "rmq-admin"
	SecretKey = "baicaf5678#"
)

var (
	// maximum waiting time for receive func
	awaitDuration = time.Second * 5
)

func main() {
	/* 日志输出至控制台 */
	os.Setenv(rmq_client.ENABLE_CONSOLE_APPENDER, "true")
	os.Setenv(rmq_client.CLIENT_LOG_LEVEL, "error")
	rmq_client.ResetLogger()

	// In most case, you don't need to create many consumers, singleton pattern is more recommended.
	pushConsumer, err := rmq_client.NewPushConsumer(&rmq_client.Config{
		Endpoint:      Endpoint,
		ConsumerGroup: ConsumerGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:    AccessKey,
			AccessSecret: SecretKey,
		},
		NameSpace: NameSpace,
	},
		rmq_client.WithPushAwaitDuration(awaitDuration),
		rmq_client.WithPushSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
			Topic: rmq_client.SUB_ALL,
		}),
		rmq_client.WithPushMessageListener(&rmq_client.FuncMessageListener{
			Consume: func(mv *rmq_client.MessageView) rmq_client.ConsumerResult {
				fmt.Println(mv)
				// ack message
				return rmq_client.SUCCESS
			},
		}),
		rmq_client.WithPushConsumptionThreadCount(20), // 消费线程
		rmq_client.WithPushMaxCacheMessageCount(10),   // 最大消息缓存数
	)
	if err != nil {
		logx.Error(err)
		return
	}
	// start pushConsumer
	if err := pushConsumer.Start(); err != nil {
		logx.Error(err)
		return
	}
	fmt.Println("Push Consumer started**********")

	sig := make(chan os.Signal)
	<-sig
	if err := pushConsumer.GracefulStop(); err != nil {
		fmt.Printf("shutdown Consumer error: %s", err.Error())
	}
}
